<?php
declare (strict_types = 1);

namespace app\controller\clean;

use app\BaseController;
use app\logic\PredisLogic;

class Tests extends BaseController
{
    private $redis;
    public function initialize()
    {
        $this->redis = new \Redis();
        $this->redis->connect(env('REDIS.HOST'), intval(env('REDIS.PORT')) );
        $this->redis->auth(env('REDIS.AUTH'));
    }

    public function index()
    {
        // 创建Redis客户端
        $redis = new PredisLogic();
        // $post_data = ['file'=>[
        //     'E003@E003@放款凭证',
        //     'Y015@7@委托保证合同',
        //     'E014@14@【数科W2】二转通知',
        //     'YE015@E015@2转收款确认函'
        // ],'type'=>[
        //     '放款凭证',
        //     '委托担保合同',
        //     '债权转让通知',
        //     '收款确认函及清单'
        // ]];
        // $insert_data = [];
        // foreach ($post_data['file'] as $key => $value) {
        //     $insert_data['ids'][] = strval(mt_rand());
        //     $insert_data['documents'][] = $value;
        //     $insert_data['metadatas'][] = ['label'=>$post_data['type'][$key]];
        // }
        // print_r($insert_data);exit;
        // Redis Stream的key
        $streamKey = 'task_stream'; // 替换为你的Stream Key
        // $data = json_encode([
        //     'id' => rand(100000,999999), // 任务ID，可以根据需要生成唯一ID
        //     'type'=>'get_file_change_name_base', // 获取知识库信息
        //     'data' => [],
        //     'callback_queue' => 'callback_queue', // 如果需要回调队列，指定队列名称
        // ],JSON_UNESCAPED_UNICODE);
        // $data = json_encode([
        //     'id' => rand(100000,999999), // 任务ID，可以根据需要生成唯一ID
        //     'type'=>'add_file_change_name_base', // 新增知识库信息
        //     'data' => $insert_data,
        //     'callback_queue' => 'callback_queue', // 如果需要回调队列，指定队列名称
        // ],JSON_UNESCAPED_UNICODE);
        
        $insert_data['ids'] = ["360110451", "1326924068", "76381899", "1394406352", "0fedb2c4ca77d2e287bd2aa4d675e66c",
		"63ec1e814e5dbb1cc30d2b19be0aff46", "ba2a42011d2f1d1ca3af34841a7910f7", "df3e784b653b5438cb5909faf4a1f0ea"];
        $data = json_encode([
            'id' => rand(100000,999999), // 任务ID，可以根据需要生成唯一ID
            'type'=>'delete_file_change_name_base', // 删除知识库信息
            'data' => $insert_data,
            'callback_queue' => 'callback_queue', // 如果需要回调队列，指定队列名称
        ],JSON_UNESCAPED_UNICODE);
        $message = ['data'=> $data];
        // 将消息发送到Redis Stream
        try {
            $result = $redis->xadd($streamKey, '*', $message);
            echo "消息已发送到Stream，消息ID: " . $result . "\n";
        } catch (Exception $e) {
            echo "发送消息失败: " . $e->getMessage() . "\n";
        }
    }
    
    public function callbackSream()
    {
        $redis = new PredisLogic(); // 你的 Redis 封装类
        $streamName = 'callback_queue';

        // 初始读取位置用 0-0 表示从第一条开始读取
        $lastId = '0-0';

        while (true) {
            try {
                // 使用 XREAD 读取最新消息（阻塞式，超时 0 表示无限等待）
                $callbackData = $redis->xread(
                    [$streamName => $lastId], // Stream 名称 + 起始 ID
                    1,                       // COUNT 1（每次读取1条）
                    10                        // BLOCK 0（无限阻塞等待）
                );
                if ($callbackData) {
                    // 提取消息数据（Predis 返回格式）
                    $message = $callbackData[0][1][0]; // [ [id, [field=>value]] ]
                    $messageId = $message[0];
                    $messageData = $message[1];

                    print_r($messageData[1]);
                    // 处理消息...
                    // echo "收到新消息 ID: $messageId, 内容: " . print_r(json_decode($messageData[1],true) ). "\n";

                    // 更新 lastId，避免重复读取
                    $lastId = $messageId;
                    $redis->xdel("callback_queue", $messageId); // 删除已处理的消息

                } else {
                    echo "没有新消息\n";
                    break;
                }
            } catch (Exception $e) {
                echo "读取消息失败: " . $e->getMessage() . "\n";
                sleep(1); // 避免频繁重试
            }
        }
    }

    public function addZjRenameAiStream()
    {
        // 接受 添加数据
        $post_data = ['file'=>[
            'Y011@5@平台协议'
        ],'type'=>[
            '平台协议'
        ]];
        // 将接受数据处理成插入数据库的数据 9c6a5b4c994c7ffdfcbfd4e5d2939a93
        $insert_data = [];
        foreach ($post_data['file'] as $key => $value) {
            $insert_data['ids'][] = getGuid();
            $insert_data['documents'][] = $value;
            $insert_data['metadatas'][] = ['label'=>$post_data['type'][$key]];
        }
        // 将数据存入数据库 

        // 存入数据库成功后，将数据发送到Redis Stream
        
        $streamKey = 'test_task_stream'; // Redis Stream的key
        $data = json_encode([
            'id' => getGuid(), // 任务ID，可以根据需要生成唯一ID
            'type'=>'add_file_change_name_base', // 新增知识库信息
            'data' => $insert_data
        ],JSON_UNESCAPED_UNICODE);
        $message = ['data'=> $data];
        // 将消息发送到Redis Stream
        try {
            $result = $this->redis->xadd($streamKey, '*', $message);
            // 发送成功给前端反馈
            echo "消息已发送到Stream，消息ID: " . $result . "\n";
        } catch (Exception $e) {
            echo "发送消息失败: " . $e->getMessage() . "\n";
        }
    }

    public function deleteZjRenameAiStream()
    {
        $streamKey = 'test_task_stream'; // Redis Stream的key
         $insert_data['ids'] = ["14f778fae25e1b4e759fc92c46ffbc5d","0fffab6003f17a082d1d4023247bf509",'2e1904836432a2b7ce4fe4dcc0e27585'];
        $data = json_encode([
            'id' => rand(100000,999999), // 任务ID，可以根据需要生成唯一ID
            'type'=>'delete_file_change_name_base', // 删除知识库信息
            'data' => $insert_data,
            'callback_queue' => 'test_callback_queue', // 如果需要回调队列，指定队列名称
        ],JSON_UNESCAPED_UNICODE);
        $message = ['data'=> $data];
        // 将消息发送到Redis Stream
        try {
            $result = $this->redis->xadd($streamKey, '*', $message);
            echo "消息已发送到Stream，消息ID: " . $result . "\n";
        } catch (Exception $e) {
            echo "发送消息失败: " . $e->getMessage() . "\n";
        }
    }

    public function getZjRenameAiStream()
    {
        $streamKey = 'test_task_stream'; // Stream Key
        $data = json_encode([
            'id' => rand(100000,999999), // 任务ID，可以根据需要生成唯一ID
            'type'=>'get_file_change_name_base', // 获取知识库信息
            'data' => [],
            'callback_queue' => 'test_callback_queue', // 如果需要回调队列，指定队列名称
        ],JSON_UNESCAPED_UNICODE);
        $message = ['data'=> $data];
        // 将消息发送到Redis Stream
        try {
            $result = $this->redis->xadd($streamKey, '*', $message);
            echo "消息已发送到Stream，消息ID: " . $result . "\n";
        } catch (Exception $e) {
            echo "发送消息失败: " . $e->getMessage() . "\n";
        }
    }

    public function callbackStream()
    {
        $streamName = 'test_callback_queue'; // 回调队列名称

        // 初始读取位置用 0-0 表示从第一条开始读取
        $lastId = '0-0';

        while (true) {
            try {
                // 使用 XREAD 读取最新消息（阻塞式，超时 0 表示无限等待）
                $callbackData = $this->redis->xread(
                    [$streamName => $lastId], // Stream 名称 + 起始 ID
                    1,                       // COUNT 1（每次读取1条）
                    10                        // BLOCK 0（无限阻塞等待）
                );
                
                if (isset($callbackData[$streamName])) {
                    foreach ($callbackData[$streamName] as $key => $value) {
                        $messageId = $key;
                        $messageData = $value['message'];
                        // 处理消息...
                        echo "收到新消息 ID: $messageId, 内容: " . $messageData. "\n";

                        // 更新 lastId，避免重复读取
                        $lastId = $messageId;
                        $this->redis->xdel($streamName, [$messageId]); // 删除已处理的消息
                    }
                } else {
                    echo "没有新消息\n";
                    break;
                }
            } catch (Exception $e) {
                echo "读取消息失败: " . $e->getMessage() . "\n";
                sleep(1); // 避免频繁重试
            }
        }
    }

    
}